package cn.doit.sparkcore2json

import java.util.Properties

import cn.doit.cn.doit.dolphinBean.{AreaTargetCountBean, MediaAnalyBean}
import cn.doit.config.JedisConfig
import com.typesafe.config.ConfigFactory
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * Created by ZHAOXUHUA 
  * .           on 2018/11/30.
  */
object MediaAnaly {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    //操作DataFrame必须有一个SQLContext的实例
    val sqlContext = new SQLContext(sc)
    val file = sqlContext.read.parquet("D:/学习专用/项目/项目四资料/data/dophinData")
    val file1 = sc.textFile("appname")
    val fileData = file.map(t => {
      //      9 requestmode: Int, 数据请求方式（1:请求、2:展示、3:点击）
      //      36 processnode: Int, 流程节点（1：请求量 kpi 2：有效请求 3：广告请
      //      31 iseffective: Int, 有效标识（有效指可以正常计费的）(0：无效 1：
      //      32 isbilling: Int, 是否收费（0：未收费 1：已收费）
      //      40 isbid: Int, 是否 rtb
      //      43 iswin: Int, 是否竞价成功
      //      42 winprice: Double, rtb 竞价成功价格
      //      76 adpayment: Double, 转换后的广告消费
      //      3 adorderid: Int, 广告 id
      //      15 appname: String, 应用名称
      //14 appid: String, 应用 id
      //      12 requestdate: String, 请求时间,格式为：yyyy-m-dd hh:mm:ss
      //      substring(requestdate, 0, 10)
      // val value = t.get(0)
      val requestmode = t.getAs[Int](8)
      val processnode = t.getAs[Int](35)
      val iseffective = t.getAs[Int](30)
      val isbilling = t.getAs[Int](31)
      val isbid = t.getAs[Int](39)
      val iswin = t.getAs[Int](42)
      val adorderid = t.getAs[Int](2)
      //分割出天
      val day = t.getAs[String](11).substring(0, 10)
      //appname
      var appname = t.getAs[String](14)
      val appid = t.getAs[String](13)
      if (appname == null) {
        if (appid != null) {
          file1.map(w_appid => {
            //val split = t.split(",")
            //val w_appname = split(0)
            //val w_appid = split(1)
            if (appid == w_appid) {
              appname = w_appid
            }
          })
        }
      }

      //原始请求
      val originalReq = if (1.equals(requestmode) && processnode >= 1) {
        1
      } else {
        0
      }
      //有效请求validReq
      val validReq = if (1.equals(requestmode) && processnode >= 2) {
        1
      } else {
        0
      }
      //广告请求advReq
      val advReq = if (1.equals(requestmode) && processnode == 3) {
        1
      } else {
        0
      }
      //参与竞价partiBid
      val partiBid = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid) && !0.equals(adorderid)) {
        1
      } else {
        0
      }
      //竞价成功succBid
      val succBid = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(iswin)) {
        1
      } else {
        0
      }
      //展示showT
      val showT = if (2.equals(requestmode) && 1.equals(iseffective)) {
        1
      } else {
        0
      }
      //点击cliT
      val cliT = if (3.equals(requestmode) && 1.equals(iseffective)) {
        1
      } else {
        0
      }
      //广告消费dspAdvConsume
      var dspAdvConsume = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid)) {
        t.getAs[Double](41) / 1000

      } else {
        0d
      }
      //广告成本dspAdvCost
      var dspAdvCost = if (1.equals(iseffective) && 1.equals(isbilling) && 1.equals(isbid)) {
        t.getAs[Double](75) / 1000
      } else {
        0d
      }

      ((day, appname), List(originalReq, validReq, advReq, partiBid, succBid, showT, cliT, dspAdvConsume, dspAdvCost))
    })

    //最后结果数据 ((天,appname), (原始请求,有效请求, 广告请求, 参与竞价 ,竞价成功 ,广告展示, 广告点击 ,广告消费, 广告成本))
    val tableData = fileData.reduceByKey(
      (t1, t2) => (
        t1.zip(t2).map(tp => tp._1 + tp._2) //zip 是对List集合进行操作
        )).map { case ((day, appname),
    List(originalReq, validReq, advReq,
    partiBid, succBid, showT, cliT,
    dspAdvConsume, dspAdvCost)) =>
      MediaAnalyBean(
        day, appname,
        originalReq.toInt, validReq.toInt, advReq.toInt,
        partiBid.toInt, succBid.toInt, showT.toInt, cliT.toInt,
        dspAdvConsume.toDouble, dspAdvCost.toDouble
      )
    }

    //写入到mysql中
    import sqlContext.implicits._

    val f = tableData.toDF()
    //写入到mysql中
    val config = ConfigFactory.load()
    val url = config.getString("dolphin.mysql.url")
    val conn = new Properties()
    conn.setProperty("user", config.getString("dolphin.mysql.username"))
    conn.setProperty("password", config.getString("dolphin.mysql.password"))
    conn.setProperty("driver", config.getString("dolphin.mysql.driver"))

    f.write.jdbc(url, "dolphin_table_M5", conn)



    sc.stop()
  }
}
